package tk.xboot.flink.vs;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;


/**
 *
 * 假设一段数据流格式长这样<1,3><1,2><1,3><2,3><2,5>
 * 那么我想对相同第一个元素所有tuple，求第二个元素的平均值。该如何实现？
 * 你可能会想到使用Flink自带的聚合函数，其中该函数缓存所有的相同key的元素，在函数里做遍历累加求值的操作。这很正确。
 * 但有一个不好的点，需要缓存所有数据。
 *
 * 如果现在就让你用map操作实现呢？而且不缓存所以数据
 * 这就需要用到状态了。试想一下，如果在map算子里面维护这样一个变量<a,b>。a是该算子的key的次数，
 * 上面数据key为1的次数便是3（a=3）,b是所有第二个元素之和。
 * 那么上面数据流在每个map算子中维护了<3,8>,<2,8>的状态。
 *
 * 好了，平均值就出来了。而且，这个状态，来一次数据更新一次，不需要缓存。
 *
 * 链接：https://www.jianshu.com/p/90ec6a76642d
 */
public class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long,Long>,Tuple2<Long,Long>> {

    /**
     *  The ValueState handle. the first field is the field, the second field is a running sum.
     */
    private transient ValueState<Tuple2<Long,Long>> sum;



    @Override
    public void flatMap(Tuple2<Long, Long> input, Collector<Tuple2<Long, Long>> out) throws Exception {

        // access the state value.
        Tuple2<Long, Long> currentSum = sum.value();

        // update the count
        currentSum.f0 ++;

        // add the second field of the input value.

        currentSum.f1 += input.f1;

        // update the state
        sum.update(currentSum);

        // if the count reaches 2, emit the average and clear the state.

        if (currentSum.f0 >= 2) {
            out.collect(new Tuple2<>(input.f0, currentSum.f1 / currentSum.f0));
            sum.clear();
        }
    }

    @Override
    public void open(Configuration conf) throws Exception {
        ValueStateDescriptor<Tuple2<Long, Long>> descriptor =
                new ValueStateDescriptor<>(
                        "average",  // the state name
                        TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {}),  // type information
                        Tuple2.of(0L,0L));  // default value of the state, if nothing is set.
        sum = getRuntimeContext().getState(descriptor);
    }
}


// this can be used in a streaming program like this (assuming we have a StreamExecutionEnvironment env)

/*
env.fromElements(Tuple2.of(1L, 3L), Tuple2.of(1L, 5L), Tuple2.of(1L, 7L), Tuple2.of(1L, 4L), Tuple2.of(1L, 2L))
        .keyBy(0)
        .flatMap(new CountWindowAverage())
        .print();
*/

// the printed output will be (1,4) and (1,5)